
import java.io.*;
import java.net.*;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.*;

public class Sender extends Thread{
	
	public ArrayList<MachineData> mData;
	public String ip;
	public int port;
	public DatagramSocket clientSocket;
	public ArrayBlockingQueue<MachineData> queue;
	public int index;
	
	public ArrayList<MachineData> send_p; //stores processers that we send heartbeats to
	public ArrayList<MachineData> receive_p; //stores processors that we expect to receive them from
	
	public ArrayList<MachineData> other; //list of processors being helped
	public ArrayList<Integer> asked;
	
	public Sender(ArrayList<MachineData> mData, int m, DatagramSocket clientSocket, ArrayBlockingQueue<MachineData> queue) {
		this.mData = mData;
		this.index = m;
		this.ip = mData.get(m).name;
		this.port = mData.get(m).port;
		this.clientSocket = clientSocket;
		this.queue = queue;
		
		send_p = new ArrayList<MachineData>();
		receive_p = new ArrayList<MachineData>();
		other = new ArrayList<MachineData>();
		asked = new ArrayList<Integer>();
		
		//assign nodes to be watched by machine and nodes watching it
		if(mData.size() > 1) { // more than one machine
			boolean done = false;
			
			for(int i = m + 1; i <= m + Global.k; i++) { //nodes to receive heartbeats
				
				int index = i % mData.size();
				
				if(index != m && !done)
					send_p.add(mData.get(index));
				else
					done =  true;
			}
			
			done = false;
			for(int i = 1; (i <= Global.k && !done); i++) { //nodes to send heartbeats
				
				int index = m - i;
				
				if (index < 0) {
					index = mData.size() + index;
				}
				
				if(index != m)
					receive_p.add(mData.get(index));
				else
					done =  true;
			}
		}
	}

	public void run() {
				
		long t = System.currentTimeMillis();
		long next = t;
		ArrayList<MachineData> currT = new ArrayList<MachineData>();
		ArrayList<MachineData> prevT = new ArrayList<MachineData>();
		
		while(true) {
			t = System.currentTimeMillis();
			
			if (t >= next) { //every timeout send ping to nodes we are monitoring
				
				queue.clear(); //clear queue because we are starting over
				asked.clear();
				
				next = t + Global.T; // we will do this again next timeout
				
				prevT = currT;
				currT = new ArrayList<MachineData>();
				
				for(MachineData m : prevT) { // if a machine hasn't not responded, consider it failed if 
					if(!m.responded) {       
						System.out.println("The processor at ip: " + m.name +" and port " + m.port + " has failed!" +
							" Detected by processor at ip " + this.ip + " and port " + this.port + ".");
						Global.addFailure();
					}
					
					for(int i = 0; i < other.size(); i++) { //remove processors we are helping
						if(mData.get(other.get(i).help_msg[1]).equals(m)) {
							other.remove(i);
						}
					}
				}
				
				try {
					Thread.sleep(100);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				
				for(MachineData m : send_p) { //send initial pings
					send(m.ip, m.port, "GIVE_A_RESPONSE", null);
				}
				
			
				for(MachineData m : receive_p) { //create new list of objects to monitor
					currT.add(new MachineData(m, t));
				}
			}
			
			// check blocking queue for messages			
			if(!queue.isEmpty()) {
				MachineData m = null;
				
				try {
					m = queue.take();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				
				if(m != null) {
					//System.out.println(m.toString());

					if(m.msg == 0x00) { //we got a response

						for(MachineData d: currT) {
							if(d.equals(m)) {
								d.responded = true;
								//System.out.println("Match: " + d.toString() + ", From: " + ip + " " + port);
								
								for(int i = 0 ; i < other.size(); i++) {
									if(mData.get(other.get(i).help_msg[1]).equals(m)) { //need to send ping letting them know m is ok
										send(other.get(i).ip, other.get(i).port,"IT_IS_OK",m);
									}
								}
							}
						}
					}
					else if(m.msg == 0x01) { // a request for message
						send(m.ip, m.port, "GIVE_A_RESPONSE", null);
					}
					else if(m.msg == 0x02) { //request to send data if hear anything
						other.add(m);
					}
					else if(m.msg == 0x03) { //got ping saying processor is ok
						int id = m.help_msg[1];//get processor id
						for(int i = 0; i < currT.size(); i++) {
							if(currT.get(i).equals(mData.get(id))) {
								currT.get(i).responded = true;
							}
						}
					}
				}
			}

			// if machine hasn't responded for roundTrip since lastPing, ping again
			for(MachineData m : currT) {
				t = System.currentTimeMillis();
				
				if(t - m.lastPing >= Global.roundTrip && !m.responded) {
					send(m.ip, m.port, "ASK_FOR_RESPONSE", null);
					//System.out.println("Haven't heard from: " + m.toString() + ", From: " + this.ip + " " + this.port);
					m.lastPing = t;
					
					//Also ask for help
					int idx = getIndex(m);
					
					boolean ask = false;
					
					for(Integer d: asked) {
						if(d == idx) {
							ask = true;
						}
					}
					
					if(!ask) {
						int i = getMachine(m);
						send(mData.get(i).ip, mData.get(i).port, "ASK_FOR_HELP", m);
						asked.add(idx);
					}
				}
			}
		}
	}
	
	public void send(InetAddress IPAddress, int port, String option, MachineData m) {
		
		if(this.index == 3) {
			if(Math.random() < 0.75)
				return;
		}
		
		byte[] sendData = new byte[1];
		
		if(option.equals("GIVE_A_RESPONSE")) //option 1 is send heartbeat
			sendData[0] = 0x00;
		else if(option.equals("ASK_FOR_RESPONSE"))//option 2 is to send response or to send reminder
			sendData[0] = 0x01;
		else if(option.equals("ASK_FOR_HELP")) { //option 3 is to ask for help
			sendData = new byte[2]; 
			sendData[0] = 0x02;
			
			int idx = getIndex(m);
			
			sendData[1] = (byte) idx;
		}
		else if(option.equals("IT_IS_OK")) {//send message to helper proc saying things are ok
			sendData = new byte[2];
			sendData[0] = 0x03;
			
			int idx = getIndex(m);
			
			sendData[1] = (byte) idx;
		}
		DatagramPacket sendPacket = new DatagramPacket(sendData, sendData.length, IPAddress, port);

		DateFormat dateFormat = new SimpleDateFormat("HH:mm:ss:SS");
		Calendar cal = Calendar.getInstance();
		System.out.println(dateFormat.format(cal.getTime())+ ": From " + this.port + " to " + port + " " + sendData[0]);
		
		try {
			clientSocket.send(sendPacket);
		}
		catch(IOException e) {
			e.printStackTrace();
		}
		
		Global.update();
	}
	
	public int getIndex(MachineData m) {
		int index = -1;
		
		for(int i = 0; i < mData.size(); i++) {
			MachineData d = mData.get(i);
			
			if(d.equals(m)) {
				index = i ;
			}
		}
		
		return index;
	}
	
	//find another machine in the list probing the possibly faulty node
	public int getMachine(MachineData m) {
				
		int index = getIndex(m);
		
		for(int i = 1; i <= Global.k; i++) {
			int temp = (index + i) % mData.size();
			
			if( temp != this.index) {
				return temp;
			}
		}
		
		return -1;
	}

}
